package dao

import (
	"context"
	"time"

	pb "bmember/api/pb/server"
	"bmember/pkg/util"

	"bmember/internal/model"
)

// MessageDao store dao
type MessageDao interface {
	// CreateMessage 创建消息
	CreateMessage(ctx context.Context, msg *model.Message) (int64, error)
	// SendMessage 发送消息
	SendMessage(ctx context.Context, msg *model.Message) error
	// GetMessageById 根据 id 获取 message
	GetMessageById(ctx context.Context, id int64) (*model.Message, error)
	// LockMessageById 根据 id 获取 message
	LockMessageById(ctx context.Context, id int64) (*model.Message, error)
	// ReSendMessageById 重试消息
	ReSendMessageById(ctx context.Context, id int64) error
	// UpdateMessageById 更新 message
	UpdateMessageById(ctx context.Context, msg *model.Message) error
	// CanMsgBeConsumed message consume check
	CanMsgBeConsumed(ctx context.Context, digest string) bool
	// RollbackCacheConsumedMsg 缓存已经消费的消息
	RollbackCacheConsumedMsg(ctx context.Context, digest string) error
	// MailMsgCache 邮件消息缓存
	MailMsgCache(ctx context.Context, digest string) bool

	// 业务相关

	// SendMemberRegisterNotify 新增用户消息
	SendMemberRegisterNotify(ctx context.Context, mem *model.Member) error
	// SendCardSendingNotify 会员领卡消息
	SendCardSendingNotify(ctx context.Context, acc *model.MemberAccount) error
}

var msgDao MessageDao

type msgDaoImpl struct {
	*DaoBase
	name string
}

// InitMessageDao 初始化 dao
func InitMessageDao() {
	msgDao = &msgDaoImpl{
		DaoBase: daoBase,
		name:    "message_dao",
	}
}

// GetMessageDao 获取 dao
func GetMessageDao() MessageDao {
	return msgDao
}

// CreateMessage 创建消息
func (dao *msgDaoImpl) CreateMessage(ctx context.Context, msg *model.Message) (int64, error) {
	if msg == nil {
		return 0, nil
	}
	param := []interface{}{
		msg.Topic,
		msg.Content,
		msg.Status,
		msg.MsgType,
		msg.CreatedAt,
		msg.UpdatedAt,
		msg.Eta,
		msg.RetryCount,
	}
	query := "INSERT INTO message (topic, content, status, msg_type, created_at, updated_at, eta, retry_count) VALUES (?,?,?,?,?,?,?,?)"
	aff, err := dao.db.PgExec(ctx, "id", query, param...)
	if err != nil {
		return 0, err
	}
	lastId, _ := aff.GetLastId()
	return lastId, nil
}

// SendMessage 发送消息
func (dao *msgDaoImpl) SendMessage(_ context.Context, msg *model.Message) error {
	if msg == nil {
		return nil
	}
	err := dao.nsq.Publish(msg.Topic, msg.Content)
	if err != nil {
		msg.NeedRetry()
	} else {
		msg.Success()
	}
	return err
}

// GetMessageById 根据 id 获取 message
func (dao *msgDaoImpl) GetMessageById(ctx context.Context, id int64) (*model.Message, error) {
	msg := &model.Message{}
	query := "SELECT * FROM message WHERE id=?"
	err := dao.db.Get(ctx, msg, query, id)
	return msg, err
}

// LockMessageById 根据 id 锁定 message
func (dao *msgDaoImpl) LockMessageById(ctx context.Context, id int64) (*model.Message, error) {
	msg := &model.Message{}
	query := "SELECT * FROM message WHERE id=? FOR UPDATE"
	err := dao.db.Get(ctx, msg, query, id)
	return msg, err
}

// UpdateMessageById 更新 message
func (dao *msgDaoImpl) UpdateMessageById(ctx context.Context, msg *model.Message) error {
	if msg == nil {
		return nil
	}
	param := []interface{}{
		msg.Status,
		msg.Eta,
		msg.RetryCount,
		msg.UpdatedAt,
		msg.Id,
	}
	query := "UPDATE message SET status=?, eta=?, retry_count=?, updated_at=? WHERE id=?"
	_, err := dao.db.Update(ctx, query, param...)
	return err
}

// ReSendMessageById 重试消息
func (dao *msgDaoImpl) ReSendMessageById(ctx context.Context, id int64) error {
	msg, err := dao.GetMessageById(ctx, id)
	if err != nil {
		return err
	}
	err = dao.SendMessage(ctx, msg)
	if err != nil {
		msg.NeedRetry()
	} else {
		msg.Success()
	}
	return dao.UpdateMessageById(ctx, msg)
}

// CanMsgBeConsumed message consume check
func (dao *msgDaoImpl) CanMsgBeConsumed(ctx context.Context, digest string) bool {
	key := dao.buildCacheKey("consumed_msg", digest)
	ok, err := dao.cache.SetNX(ctx, key, "ok", 1*time.Hour)
	if err != nil {
		return true
	}
	return ok
}

// RollbackCacheConsumedMsg 缓存已经消费的消息
func (dao *msgDaoImpl) RollbackCacheConsumedMsg(ctx context.Context, digest string) error {
	key := dao.buildCacheKey("consumed_msg", digest)
	_, err := dao.cache.Del(ctx, key)
	return err
}

// MailMsgCache 邮件消息缓存
func (dao *msgDaoImpl) MailMsgCache(ctx context.Context, digest string) bool {
	key := dao.buildCacheKey("sent_mail", digest)
	ok, err := dao.cache.SetNX(ctx, key, "ok", 10*time.Minute)
	if err != nil {
		return true
	}
	return ok
}

// SendMemberRegisterNotify 新增用户消息
func (dao *msgDaoImpl) SendMemberRegisterNotify(ctx context.Context, mem *model.Member) error {
	if mem == nil {
		return nil
	}
	payload := &pb.MemberRegisterNotify{
		MemberId:  mem.Id,
		BrandId:   mem.BrandId,
		Birthday:  util.TimeToDate(mem.Birthday),
		Avatar:    mem.Avatar,
		Mobile:    mem.Mobile,
		Email:     mem.Email,
		Nickname:  mem.Nickname,
		Timestamp: util.GetCurrentTimestamp(),
	}
	msg := model.NewMessage(dao.cfg.Topics.TopicMemberRegister, payload, model.MsgTypeForSending)
	err := dao.SendMessage(ctx, msg)
	_, _ = dao.CreateMessage(ctx, msg)
	return err
}

// SendCardSendingNotify 会员领卡消息
func (dao *msgDaoImpl) SendCardSendingNotify(ctx context.Context, acc *model.MemberAccount) error {
	if acc == nil {
		return nil
	}
	payload := &pb.CardSendingNotify{
		AccId:     acc.Id,
		MemberId:  acc.MemberId,
		GroupId:   acc.GroupId,
		CardId:    acc.CardId,
		BrandId:   acc.BrandId,
		Balance:   acc.Balance,
		Integral:  acc.Integral,
		Timestamp: util.GetCurrentTimestamp(),
	}
	msg := model.NewMessage(dao.cfg.Topics.TopicCardSend, payload, model.MsgTypeForSending)
	err := dao.SendMessage(ctx, msg)
	_, _ = dao.CreateMessage(ctx, msg)
	return err
}
